package com.atbeijing.D03;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 不分流直接开窗
 */
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
            .addSource(new SensorSource())
            //机器时间的滚动窗口,为整个流开窗,基本上没啥用
            .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .reduce(new ReduceFunction<SensorReading>() {
                @Override
                public SensorReading reduce(SensorReading value1, SensorReading value2) throws Exception {
                    return new SensorReading("sensor",value1.temperature + value2.temperature,0L);
                }
            })
            .print();

        env.execute();
    }
}
